package com.demo.springbootdemonettywebsocket.socket;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.demo.springbootdemonettywebsocket.pojo.MessagePojo;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.MultiValueMap;
import org.yeauty.annotation.*;
import org.yeauty.pojo.Session;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * bossLoopGroupThreads建议设置1
 * workerLoopGroupThreads建议设置为CPU*2
 * 另外需要查看服务器的文件打开数是否有限制
 *
 * @author peter
 * @date 2022/5/18 上午8:37
 */
@ServerEndpoint(path = "/ws/connect", port = "${ws.port}", maxFramePayloadLength = "131070", childOptionWriteBufferHighWaterMark = "131070",readerIdleTimeSeconds = "0")
@Slf4j
public class ConnectWebSocket {

    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static AtomicInteger onlineCount = new AtomicInteger(0);
    private static ConcurrentHashMap<String, Session> webSocketClientMaps = new ConcurrentHashMap<String, Session>();

    private Timer timer;

    //心跳检验
    private volatile boolean isPong;

    @BeforeHandshake
    public void handshake(Session session, HttpHeaders headers, @RequestParam String req, @RequestParam MultiValueMap reqMap, @PathVariable String arg, @PathVariable Map pathMap){
        session.setSubprotocols("stomp");

        // protobuf
        // 在线网站测试
        // http://www.websocket-test.com/
        // Socket地址 ws://127.0.0.1:18009/ws/connect?req=ok&username=peter
        // 其中req=ok是必填

        // 前端测试 console
        // socket = new WebSocket("ws://127.0.0.1:18009/ws/connect?req=ok&username=peter");
        log.info("req->" + req);// req->ok
        log.info("reqMap->" + reqMap.toString());
        log.info("arg->" + arg);
        log.info("pathMap->" + pathMap.toString());

        if (!"ok".equals(req)){
            log.warn("Authentication failed!");
            session.close();
        }
    }

    /**
     * 当有新的WebSocket连接进入时，对该方法进行回调 注入参数的类型:Session、HttpHeaders、ParameterMap
     * @param session
     * @param headers
     * @param username
     * @param reqMap
     * @param arg
     * @param pathMap
     */
    @OnOpen
    public void onOpen(Session session, HttpHeaders headers, @RequestParam String username, @RequestParam MultiValueMap reqMap, @PathVariable String arg, @PathVariable Map pathMap) {
        int i = onlineCount.incrementAndGet();
        log.info("onOpen {} -> {}, session.id[{}]上线, 在线连接数[{}], 当前用户username [{}]",
                session.channel().remoteAddress().toString(),
                session.channel().localAddress().toString(),
                session.id(),
                String.valueOf(i),
                username
        );
    }


    /**
     * 当有WebSocket连接关闭时，对该方法进行回调 注入参数的类型:Session
     *
     * @param session
     * @throws IOException
     */
    @OnClose
    public void onClose(Session session) throws IOException {
        int i = onlineCount.decrementAndGet();
        log.info("OnClose -> [" + session.id() + "]下线, 在线连接数" +String.valueOf(i));
    }

    /**
     * 当有WebSocket抛出异常时，对该方法进行回调 注入参数的类型:Session、Throwable
     * @param session
     * @param throwable
     */
    @OnError
    public void onError(Session session, Throwable throwable) {
        throwable.printStackTrace();
    }

    /**
     * 当接收到字符串消息时，对该方法进行回调 注入参数的类型:Session、String
     * @param session 会话信息
     * @param message 消息内容
     */
    @OnMessage
    public void onMessage(Session session, String message) {

        log.info("onMessage.session->" + session.toString());

        if ("ping".equals(message.toLowerCase().toString())) {
            // 回复pong
            session.sendText("pong");
        } else {
            AttributeKey<String> key = AttributeKey.valueOf("message");

            String oldMessage = session.channel().attr(key).getAndSet(message);
            if (! StrUtil.isEmpty(oldMessage)) {
                session.sendText(StrUtil.format("上一条信息[{}], 最新信息[{}]", oldMessage, message));
            }
        }

    }

    @OnBinary
    public void onBinary(Session session, byte[] bytes) {
        for (byte b : bytes) {
            System.out.println(b);
        }
        session.sendBinary(bytes);
    }

    /**
     * 当接收到Netty的事件时，对该方法进行回调 注入参数的类型:Session、Object
     * @param session
     * @param evt
     */
    @OnEvent
    public void onEvent(Session session, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
            switch (idleStateEvent.state()) {
                case READER_IDLE:
                    System.out.println("read idle");
                    if (session.isOpen()) {
                        session.sendText("PING");
                    }
                    break;
                case WRITER_IDLE:
                    System.out.println("write idle");
                    if (session.isOpen()) {
                        session.sendText("PING");
                    }
                    break;
                case ALL_IDLE:
                    System.out.println("all idle");
                    if (session.isOpen()) {
                        session.sendText("PING");
                    }
                    break;
                default:
                    break;
            }
        }
    }

}
